/*********************************************************************************
 *      Copyright:  (C) 2022 Zhang Changxing<ZhangChangxingVIP@163.com>
 *                  All rights reserved.
 *
 *       Filename:  mosq.c
 *    Description:  This file is used to connect MQTT, publish and subscribe topics.
 *                 
 *        Version:  1.0.0(05/1/2022)
 *         Author:  Zhang Changxing <ZhangChangxingVIP@163.com>
 *      ChangeLog:  1, Release initial version on "05/1/2022  01:33:25 AM"
 *                 
 ********************************************************************************/

#include "mosq.h"

struct mosquitto *mosquitto_ctx = NULL; 
char *shared_buf[1024];
pthread_attr_t thread_attr;




void connect_callback(struct mosquitto *mosq, void *obj, int rc)
{
    int recon_rc;

    if(rc != 0)
    {
        log_error("Connect server failure,rc= %d\n", rc);
        recon_rc = mosquitto_reconnect(mosq);
        if(MOSQ_ERR_SUCCESS == recon_rc)
        {
            log_info("Reconnect broker sucessfully!\n");
        }

        return;
    }
}

void disconnect_callback(struct mosquitto *mosq,void *obj, int rc)
{
    if(rc == 0)
    {
        log_warn("The client called the mosquitto_disconnect function!\n");
        return;
    }

}

void publish_callback(struct mosquitto *mosq, void *obj, int mid)
{
    pub_ctx_t *ptr = (pub_ctx_t *)obj;
    ptr->pub_flag = 1;
    ptr->connect_flag = 1;
    log_info("Message id:%d publish sucessfully!\n",mid);
    log_debug("pub_flag:%s\n", ptr->pub_flag);
    log_info("Message:%s\n", ptr->pub_message);
    return;
}

void subscribe_callback(struct mosquitto *mosq, void *obj, int mid, int qos_count, const int *granted_qos)
{
    log_debug("The proxy responds to a subscription !\n");
}

void message_callback(struct mosquitto *mosq, void *obj, const struct mosquitto_message *message)
{
    // pub_ctx_t *ptr = (pub_ctx_t *)obj;
    // ptr->pub_flag = 1;
    // ptr->connect_flag = 1;
    // //log_info("The agent confirms that it successfully received the following message:\n");
    // //log_info("mid:%d\ntopic:%s\npayload:%s\nqos:%d\n", message->mid, message->topic, (char*)message->payload, message->qos);
     return;
}

void log_callback(struct mosquitto *mosq, void *obj, int level, const char *str)
{
    log_debug("log_callback:%s\n", str);
}



/*--------------------------------------------------------------------------------------------------
 *   description:   Connect to brokers using the MQTT protocol
 *          args:   $mqtt_connect_ctx:    The configuration information structure pointer that connects the broker
 *  return value:      <0: failure,    =0:ok 
 *--------------------------------------------------------------------------------------------------*/
int mqtt_init(iotd_ctx_t *iotd_ctx)
{
    int rv = -1;
    int session = 1;

    if(!iotd_ctx)
    {
        log_error("The %s() arguments input error!\n", __func__);
        return -1;
    }

    mqtt_connect_ctx_t *mqtt_connect_ctx = &(iotd_ctx->mqtt_connect_ctx);
    pub_ctx_t   *pub_ctx = &(iotd_ctx->pub_ctx);
    sub_ctx_t   *sub_ctx = &(iotd_ctx->sub_ctx);

    rv = mosquitto_lib_init();
    if(rv != MOSQ_ERR_SUCCESS)
    {
        log_error("mosquitto init failure:%s\n", strerror(errno));
        return -2;
    }

    mosquitto_ctx = mosquitto_new(mqtt_connect_ctx->client_id, true, (void *)pub_ctx);
    if(!mosquitto_ctx)
    {
        log_error("Create a new mosquitto failure:%s\n", strerror(errno));
        return -3;
    }

    mosquitto_connect_callback_set(mosquitto_ctx, connect_callback);
    mosquitto_disconnect_callback_set(mosquitto_ctx, disconnect_callback);
    mosquitto_publish_callback_set(mosquitto_ctx, publish_callback);
    mosquitto_subscribe_callback_set(mosquitto_ctx, subscribe_callback);
    mosquitto_message_callback_set(mosquitto_ctx,message_callback);
    mosquitto_log_callback_set(mosquitto_ctx, log_callback);
    rv = mosquitto_username_pw_set(mosquitto_ctx, mqtt_connect_ctx->username,  mqtt_connect_ctx->passwd);
    if(rv == MOSQ_ERR_INVAL)
    {
        log_error("Invalid argument!\n");
        return -4;
    }
    else if(rv == MOSQ_ERR_NOMEM)
    {
        log_error("Out of memory!\n");
        return -5;
    }

    log_debug("Username and passwd set sucessfully!\n");
    
    log_debug("\n\n\nhost is:%s\n", mqtt_connect_ctx->host);
    log_debug("port is:%d\n", mqtt_connect_ctx->port);
    log_debug("username is:%s\n", mqtt_connect_ctx->username);
    log_debug("passwd is:%s\n", mqtt_connect_ctx->passwd);
    log_debug("client_id is:%s\n", mqtt_connect_ctx->client_id);
    log_debug("keepalive is:%d\n\n\n\n", mqtt_connect_ctx->keepalive);


    // if(subscribe_data(*sub_ctx) < 0)
    // {
    //     return -6;
    // }

    if(0 == mqtt_connect(mqtt_connect_ctx))
    {
        pub_ctx->connect_flag = 1;
        log_info("connect sucessfully!\n");
    }

    #if 1
    rv = mosquitto_loop_start(mosquitto_ctx);
    if(rv != MOSQ_ERR_SUCCESS)
    {
        log_error("mosquitto loop start error!\n");
        return -7;
    }
    #endif

    #if 1
    rv = mosq_loop_start(iotd_ctx);
    if(rv != 0)
    {
        log_error("mosq loop start error!\n");
        return -7;
    }
    #endif

    return 0;

}


/*--------------------------------------------------------------------------------------------------
 *   description:   Used to connect to MQTT
 *          args:   $mqtt_connect_ctx: The context address used to connect to MQTT
 *  return value:      <0: failure,    =0:ok 
 *--------------------------------------------------------------------------------------------------*/
int mqtt_connect(mqtt_connect_ctx_t *mqtt_connect_ctx)
{   
    int rv;

    rv = mosquitto_connect(mosquitto_ctx, mqtt_connect_ctx->host, mqtt_connect_ctx->port, mqtt_connect_ctx->keepalive);
    if(rv == MOSQ_ERR_INVAL)
    {
        log_error("mosquitto_connect(),Invalid argument!\n");
        return -1;
    }
    else if(rv == MOSQ_ERR_ERRNO)
    {
        log_error("mosqtuitto connect failure:%s\n", strerror(errno));
        return -2;
    }
    else if(rv == MOSQ_ERR_SUCCESS)
    {
        log_warn("mosqtuitto connect sucessfully:%s\n", strerror(errno));
        return 0;
    }
}


/*--------------------------------------------------------------------------------------------------
 *   description:   Used to reconnect to MQTT
 *  return value:      <0: failure,    =0:ok 
 *--------------------------------------------------------------------------------------------------*/
int mqtt_reconnect(void)
{
    int rv;

    rv = mosquitto_reconnect(mosquitto_ctx);
    if(MOSQ_ERR_SUCCESS != rv)
    {
        log_debug("Reconnect broker faile!\n");
        return -1;
    }

    log_info("Reconnect broker sucessfully!\n");
    return 0;
}


/*--------------------------------------------------------------------------------------------------
 *   description:   Sends data to connected servers
 *          args:   $pub_ctx:    The configuration information structure that connects the broker
 *  return value:      <0: failure,    =0:ok 
 *--------------------------------------------------------------------------------------------------*/
int publish_data(pub_ctx_t *pub_ctx)
{
    int rv = -1;


    rv = mosquitto_publish(mosquitto_ctx, NULL, pub_ctx->pub_topic, strlen(pub_ctx->pub_message), pub_ctx->pub_message, pub_ctx->pub_qos, true);
    if(MOSQ_ERR_SUCCESS != rv)
    {
        log_error("Client mosquitto pubscribe failure!\n");
        return -1;
    }

    return 0;
}


/*--------------------------------------------------------------------------------------------------
 *   description:   Subscribe to topic messages
 *          args:   $sub_ctx:    The configuration information structure that connects the broker
 *  return value:      <0: failure,    =0:ok 
 *--------------------------------------------------------------------------------------------------*/
int subscribe_data(sub_ctx_t sub_ctx)
{
    int mid;
    int rv = -1;

    rv = mosquitto_subscribe(mosquitto_ctx, &mid, sub_ctx.sub_topic, sub_ctx.sub_qos);
    if(MOSQ_ERR_SUCCESS != rv)
    {
        log_error("Client mosquitto subscribe failure!\n");
        return -1;
    }

    log_info("Client mosquitto subscribe sucessfully!\n");
    return 0;
}


/*--------------------------------------------------------------------------------------------------
 *   description:   Unsubscribe topic messages
 *          args:   $sub_ctx:    The configuration information structure that connects the broker
 *  return value:      <0: failure,    =0:ok 
 *--------------------------------------------------------------------------------------------------*/
int unsubscribe_data(sub_ctx_t sub_ctx)
{
    int mid;
    int rv = -1;
    
    rv = mosquitto_unsubscribe(mosquitto_ctx, &mid, sub_ctx.sub_topic);
    if(MOSQ_ERR_SUCCESS != rv)
    {
        log_error("Client mosquitto unsubscribe failure!\n");
        return -1;
    }
    log_debug("Client mosquitto unsubscribe sucessfully!\n");

}


/*--------------------------------------------------------------------------------------------------
 *   description:   When MQTT is disconnected unexpectedly, reconnect
 *          args:   $iotd_ctx: The context address used to reconnect to MQTT
 *  return value:      <0: failure,    =0:ok 
 *--------------------------------------------------------------------------------------------------*/
int mosq_loop_start(iotd_ctx_t *iotd_ctx)
{
    pthread_t   tid;

    if(!iotd_ctx)
    {
        log_error("The %s() arguments input error!\n", __func__);
        return -1;
    }

    if (pthread_attr_init(&thread_attr))
    {
        log_error("pthread_attr_init() failure:%s\n", strerror(errno));
        return -1;
    }

    if (pthread_attr_setstacksize(&thread_attr, 120*1024))
    {
        log_error("pthrea_attr_setstacksize() failure:%s\n",strerror(errno));
        return -2;
    }

    if (pthread_attr_setdetachstate(&thread_attr, PTHREAD_CREATE_DETACHED))
    {
        log_error("pthrea_attr_setdetachstate() failure:%s\n", strerror(errno));
        return -3;
    }

    printf("Thread_worker:[] start running....\n");
    pthread_create(&tid, NULL, thread_worker, iotd_ctx);
    
}


/*--------------------------------------------------------------------------------------------------
 *   description:   Creating a child thread executes this function, which is used to process messages\
                    and commands from the server
 *          args:   $args:    The configuration information structure that connects the broker
 *--------------------------------------------------------------------------------------------------*/
void *thread_worker(void *args)
{
    int rv;

    iotd_ctx_t *iotd_ctx = (iotd_ctx_t *)args;
    pub_ctx_t *pub_ctx = &(iotd_ctx->pub_ctx);
    mqtt_connect_ctx_t *mqtt_connect_ctx = &(iotd_ctx->mqtt_connect_ctx);

    if (!args)
    {
        log_error("%s() get invalid argument!\n", __FUNCTION__);
        pthread_exit(NULL);
    }

    log_info("Thread_worker:[%ld] start running....\n", pthread_self());

    //subscribe_data(*pub_ctx);

    while (1)
    {
        if(0 == pub_ctx->connect_flag)
        {
            if(0 == mqtt_reconnect())
            {
                pub_ctx->connect_flag = 1;
            }
            continue;
        }

    }


}


/*--------------------------------------------------------------------------------------------------
 *   description:   Sends data to connected servers
 *  return value:      <0: failure,    =0:ok 
 *--------------------------------------------------------------------------------------------------*/
int mosquitto_cleanup(void)
{
    int rv;

    mosquitto_destroy(mosquitto_ctx);

    pthread_attr_destroy(&thread_attr);

    rv = mosquitto_lib_cleanup();
    if(MOSQ_ERR_SUCCESS != rv)
    {
        log_warn("mosquitto_lib_cleanup() execute failure!\n");
        return -1;
    }

    log_info("mosquitto cleanup sucessfully!\n");
    
    return 0;
}



